package com.kingsoft.dc.khaos.module.spark.util

import java.util

import com.alibaba.fastjson.JSONObject
import com.kingsoft.dc.khaos.KhaosContext
import com.kingsoft.dc.khaos.extender.meta.api.{AuthRequest, DmRequest, DmTableSplit, ReportDataStatusResult}
import com.kingsoft.dc.khaos.extender.meta.model.MetaParams
import com.kingsoft.dc.khaos.extender.meta.utils.DmMetaUtils
import com.kingsoft.dc.khaos.module.spark.constants.SchedulerConstants
import org.json4s.{DefaultFormats, JValue}
import org.json4s.jackson.JsonMethods.parse

import scala.collection.mutable.HashMap

/**
  * Created by haorenhui on 2019/09/24.
  */
object DataStatusUtils {

    //上报KV 枚举
    object IndicatorsEnum extends Serializable{
        // 关系型数据库
        //当前业务时间（更新数据时间）  value: 时间格式 2019-08-14 10:15:37
        val CURRENT_BTIME  = "CURRENT_BTIME"
        //数据条数
        val DATA_NUM = "DATA_NUM"
        //最新分区  value:字符串
        val LAST_PARTION = "LAST_PARTION"
        //是否覆盖    默认false    ture则是覆盖   false  累加
        val isCover = "IS_COVER"

        // cos
        //数据大小  value:单位bit
        val DATA_SIZE = "DATA_SIZE"
        //文件路径 value: 字符串
        val FILE_LOCATION = "FILE_LOCATION"
        //文件的名称 value: 字符串
        val FILE_NAME = "FILE_NAME"
    }

    def buildReportDataStatusRequest(kc: KhaosContext): DmRequest = {
        val request = new DmRequest()
        request.setUrl(kc.conf.getString(SchedulerConstants.META_MG_SERVICE_META_MG))
        val authRequest: AuthRequest = new AuthRequest()
        authRequest.setAuthUrl(kc.conf.getString(SchedulerConstants.AUTH_SERVER_KEY))
        request.setAuthRequest(authRequest)
        request
    }

    /** 生成数据状态上报的请求json */
    def generateReportDataStatusJson(metaRequest:DmRequest,indicators:HashMap[String,String]) :String={
        val requestJObject = new JSONObject()
        val datasObjects = new util.ArrayList[JSONObject]()
        val indicatorsObjects = new util.ArrayList[JSONObject]()
        val jsonObj :JSONObject = new JSONObject()
        jsonObj.put("dsId",metaRequest.getMetaParams.getDsId)
        jsonObj.put("dsName",metaRequest.getMetaParams.getDsName)
        jsonObj.put("dbName",metaRequest.getMetaParams.getDbName)
        jsonObj.put("tlbName",metaRequest.getMetaParams.getTblName)
        indicators.foreach(kv=>{
            val key: String = kv._1
            val value: String = kv._2
            val kvObj = new JSONObject()
            kvObj.put("key",key)
            kvObj.put("value",value)
            indicatorsObjects.add(kvObj)
        })
        jsonObj.put("indicators",indicatorsObjects)
        datasObjects.add(jsonObj)

        requestJObject.put("env",metaRequest.getMetaParams.getEnv)
        requestJObject.put("datas",datasObjects)
        println("数据状态上报   "+requestJObject.toJSONString)
        requestJObject.toJSONString
    }


    /**
      * 上报数据状态
      * @param kc 上下文
      * @param dbName 数据库名称
      * @param tblName 表名称
      * @param indicatorsMap 上报基本指标的key值映射
      * @param className
      * @param metaParamsJson
      * @return com.kingsoft.dc.khaos.extender.meta.api.ReportDataStatusResult
      */
    def reportDataStatus[T](kc: KhaosContext, dbName: String, tblName: String, indicatorsMap: HashMap[String,String],className: String, metaParamsJson: String): ReportDataStatusResult = {
        val runEnv = kc.conf.getString(SchedulerConstants.RUN_ENV)

        val metaRequest = buildReportDataStatusRequest(kc)
        val metaParams = new MetaParams(metaParamsJson)
        metaParams.buildWriteDatasourceParams(runEnv, dbName, tblName)

        implicit val formats:DefaultFormats = DefaultFormats
        val value: JValue = parse(metaParamsJson, true:Boolean)
        val metaMap: Map[String, String] = value.values.asInstanceOf[Map[String, String]]
        metaParams.setDsName(metaMap.getOrElse("ds_name",null))
        metaRequest.setMetaParams(metaParams)

        val requestJson: String = generateReportDataStatusJson(metaRequest,indicatorsMap)
        metaRequest.setParams(requestJson)

        val reportDataStatusResult: ReportDataStatusResult = DmMetaUtils.reportDataStatus(className,metaRequest,classOf[ReportDataStatusResult])
        reportDataStatusResult
    }




}
